<?php
/**
 * Created by PhpStorm.
 * User: Administrator
 * Date: 2019/8/14 0014
 * Time: 下午 10:15
 */

namespace App\Listener;

use App\Rpc\Lib\NotifyInterface;
use Swoft\Bean\Annotation\Mapping\Inject;
use Swoft\Event\Annotation\Mapping\Listener;
use Swoft\Event\EventHandlerInterface;
use Swoft\Event\EventInterface;

use Swoft\Server\Swoole\SwooleEvent;

/**
 * Class MessageServer
 * @package App\Listener
 * @Listener(SwooleEvent::START)
 */
class MessageServer implements EventHandlerInterface
{

    /**
     * @\Swoft\Rpc\Client\Annotation\Mapping\Reference(pool="notify.pool")
     *
     * @var NotifyInterface
     */
    private $notifyService;

    /**
     * @Inject("rabbit.pool")
     * @var \Six\Rabbit\Pool
     */
    private $rabbit;

    public function handle(EventInterface $event): void
    {
        sgo(function (){
            swoole_timer_tick(1000,function (){
                #延迟队列
                $delayExchangeName = 'delayExchange';
                $delayRouteKey = '/delay';
                $delayQueueName = 'delay';

                //创建连接
                $connection = $this->rabbit->connect();
                $connectionRabbit = $connection->connection;
                $channel = $connectionRabbit->channel(); //建立通道
                //缓存队列
                $channel->exchange_declare($delayExchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, false, true, false);
                #声明一个队列，发送消息使用
                $channel->queue_declare($delayQueueName,false,true,false,false);
                $channel->queue_bind($delayQueueName,$delayExchangeName,$delayRouteKey);

                $channel->basic_consume($delayQueueName,'',false,false,false,false,function ($message){

                    $data = json_decode($message->body ,true);

                    var_dump($data);
                    $data['notify_retries_number']+=1;
                    if($data['notify_retries_number'] >= 3){
                        //写入redis中
                        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
                    }
//                    var_dump("执行");
                    //    system("php /data/wwwroot/rabbitmq/phprabbit/app/send.php");
                    //回应ack
                    $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
                });

                // Loop as long as the channel has callbacks registered
                while ($channel->is_consuming()) {
                    $channel->wait(); //阻塞消费
                }

                /**
                 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
                 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
                 */
                function shutdown($channel, $connection)
                {
                    $channel->close();
                    $connection->close();
                }

                register_shutdown_function('shutdown', $channel, $connectionRabbit);
            });
        });
    }
}